package com.dwk.rabbitMQ5;


import com.dwk.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 发布确认模式比较：
 * 1.单个发布确认模式
 * 2.批量确认
 * 3.异步批量确认
 */
public class ConfirmMessage {
    public static final int MESSAGE_COUNT=1000;

    public static void main(String[] args) throws Exception {
        //1.单个确认
        //ConfirmMessage.publishMessageIndividually();//36565ms

        //2.批量确认
        //ConfirmMessage.publishMessageBatch();//669ms

        //3.异步批量确认
        ConfirmMessage.publishMessageAsync();//302ms
    }

    //单个确认
    public static void publishMessageIndividually() throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        //声明队列
        String queueName= UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认模式
        channel.confirmSelect();
        //开始时间
        long begin =System.currentTimeMillis();
        //批量发消息,单个确认
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message=i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //单个消息马上进行确认
            boolean flag = channel.waitForConfirms();
            if(flag){
                System.out.println("消息确认成功");
            }
        }
        //结束时间
        long end =System.currentTimeMillis();
        System.out.println("单个发布确认模式发布1000条数据的时候，耗时："+(end-begin)+"毫秒");
    }


    //批量确认
    public static void publishMessageBatch() throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        //声明队列
        String queueName= UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认模式
        channel.confirmSelect();
        //开始时间
        long begin =System.currentTimeMillis();

        //未确认消息的个数
        int outstandingMessageCount = 0;

        //批量发消息,批量确认
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message=i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //达到100条消息的时候批量的确认一次
            outstandingMessageCount++;
            if (outstandingMessageCount == 100) {
                channel.waitForConfirms();
                outstandingMessageCount=0;
                System.out.println("确认第"+(i/100)+"次");
            }
        }
        //为了确保还有剩余没有确认消息 再次确认
        if (outstandingMessageCount > 0) {
            channel.waitForConfirms();
        }
        //结束时间
        long end =System.currentTimeMillis();
        System.out.println("批量发布确认模式发布1000条数据的时候，耗时："+(end-begin)+"毫秒");
    }

    //异步确认
    public static void publishMessageAsync() throws Exception{
        Channel channel = RabbitMqUtils.getChannel();
        //声明队列
        String queueName= UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认模式
        channel.confirmSelect();

        /**
          * 线程安全有序的一个哈希表，适用于高并发的情况下
         *  通过序号删除消息
         *  支持高并发
         */
        ConcurrentSkipListMap<Long,String> outstandingConfirms=new ConcurrentSkipListMap<>();

        //消息确认成功回调函数
        ConfirmCallback ackCallback =(deliveryTag,multiple) ->{
            //2.删除掉已经确认的消息
            if (multiple) {
                //返回的是小于等于当前序列号的未确认消息 是一个 map
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag, true);
                //清除该部分未确认消息
                confirmed.clear();
            }else{
                //只清除当前序列号的消息
                outstandingConfirms.remove(deliveryTag);
            }
            System.out.println("确认过的消息"+deliveryTag);
        };

        //消息确认失败回调函数
        ConfirmCallback nackCallback =(deliveryTag,multiple) ->{
            //3.打印未确认的消息有哪些
            String message=outstandingConfirms.get(deliveryTag);
            System.out.println("未确认的消息"+deliveryTag);
        };
        //准备一个监听器，看哪些消息成功了，哪些失败了
        channel.addConfirmListener(ackCallback,nackCallback);//异步通知

        //开始时间
        long begin =System.currentTimeMillis();

        //批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message=i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //1.此处记录下所有要发送的消息
            outstandingConfirms.put(channel.getNextPublishSeqNo(),message);
        }
        //结束时间
        long end =System.currentTimeMillis();
        System.out.println("异步发布确认模式发布1000条数据的时候，耗时："+(end-begin)+"毫秒");
    }
}
